package com.navinfo.opentsp.platform.computing.analysis.application;

import com.fasterxml.jackson.databind.JsonMappingException;
import com.navinfo.opentsp.platform.computing.analysis.entity.*;
import com.navinfo.opentsp.platform.computing.analysis.util.*;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.storage.StorageLevel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

import java.io.IOException;
import java.math.BigDecimal;
import java.sql.*;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * Created by neo on 2017/6/1.
 */
public class VehiclePassTimesStatistics {

    private static Logger logger = LoggerFactory.getLogger(VehiclePassTimesStatistics.class);

    public static void main(String args[]) throws ClassNotFoundException, InterruptedException {

        final String day = args[0];
        int num = Integer.parseInt(args[1]);
        logger.error("读取参数:,day:{},num:{}",day,num);
        String master = PropertiesUtil.getProperties("spark.master.type");
        SparkSession spark = getSparkSessionByMaster(master);
        JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());
        logger.error("[1/4]......读取数据源");
        //匹配服务站与经纬度位置关系
        AreaTileCache areaTileCache = getAreaTile();
        //读取瓦片对照省市文件路径
        String tileDB =  PropertiesUtil.getProperties("vehiclePasstimes.hive.dbname.tile");
        String tileTB =  PropertiesUtil.getProperties("vehiclePasstimes.hive.tbname.tile");
        String tileHQL =  PropertiesUtil.getProperties("vehiclePasstimes.hive.hql.tile");
        String tileSql = String.format(tileHQL,tileDB,tileTB);
        logger.error("OOO-->tile.hql:{}",tileSql);
        //读取省市区经纬度对照表文件路径
        String areaDB =  PropertiesUtil.getProperties("vehiclePasstimes.hive.dbname.area");
        String areaTB =  PropertiesUtil.getProperties("vehiclePasstimes.hive.tbname.area");
        String areaHQL =  PropertiesUtil.getProperties("vehiclePasstimes.hive.hql.area");
        String areaSql = String.format(areaHQL,areaDB,areaTB);
        logger.error("OOO-->area.hql:{}",areaSql);
        /**
         * 创建广播变量
         */
        //电子围栏服务站与瓦片信息
        Broadcast<HashMap<Long, HashSet<Long>>> mapA = jsc.broadcast(areaTileCache.getAreaTileCache());
        //瓦片信息
        Broadcast<HashMap<Long, String>> mapB = TileMap.getInstance(spark,jsc, tileSql);
        //经纬度对照省市区信息
        Broadcast<HashMap<Integer,String>> mapC = AreaMap.getInstance(spark,jsc,areaSql);

        /**
         * 读取统一协议数据
         */
        String protocolDB = PropertiesUtil.getProperties("vehiclePasstimes.hive.dbname");
        String protocolTB = PropertiesUtil.getProperties("vehiclePasstimes.hive.tbname");
        String protocolHQL = PropertiesUtil.getProperties("vehiclePasstimes.hive.hql");
        String protocolSql = String.format(protocolHQL,protocolDB,protocolTB,day);
        logger.error("OOO-->protocol.hql:{}",protocolSql);
        Dataset<Row> dataset = spark.sql(protocolSql);
        JavaRDD<Row> lines = dataset.toJavaRDD();
        JavaRDD<LocationDataInfo> data = lines.mapPartitions((FlatMapFunction<Iterator<Row>, LocationDataInfo>) rows -> new Iterator<LocationDataInfo>() {
            @Override
            public boolean hasNext() {
                return rows.hasNext();
            }

            @Override
            public LocationDataInfo next() {
                Row row = rows.next();
                LocationDataInfo locationDataInfo = VehiclePassTimesStatistics.LocationDataInfo.parseObj(row);
                return locationDataInfo;
            }
        });
        JavaRDD<LocationDataInfo> dataCache = data.persist(StorageLevel.MEMORY_AND_DISK_SER());

        //判断是否开启热力图
        boolean issaveGridData = Boolean.parseBoolean(
                Optional.ofNullable(
                        PropertiesUtil.getProperties("save.vehicleDrivingNumber.flag"))
                        .orElse("false"));
        logger.error("[2/4]......热力图相关计算开始，是否计算:{}",issaveGridData);
        if(issaveGridData) {
            //热力图1

            JavaRDD<VehicleDrivingNumberInGridDto> cityGridRDD = getCityGirdRDD(dataCache, num, day, mapB, mapC);

            //热力图2
            JavaRDD<VehicleDrivingNumberInGridDto> provinceGridRDD = getProvinceGirdRDD(dataCache, num, day, mapB, mapC);
            //合并热力图RDD，统一落盘
            JavaRDD<VehicleDrivingNumberInGridDto> areaGridRDD = cityGridRDD.union(provinceGridRDD);

            //缓存热力图数据RDD，需要落盘两次，且逻辑较复杂
            JavaRDD<VehicleDrivingNumberInGridDto> areaGridRDDCache = areaGridRDD.persist(StorageLevel.MEMORY_AND_DISK_SER());

            saveVehicleDrivingNumberInGridDtoByTypeList(spark,areaGridRDDCache,day);
            areaGridRDDCache.unpersist();
        }
        //关联车辆信息和位置信息
        JavaPairRDD<String,String> mapRDD = getVehicleCrossArea(dataCache,mapA,mapB,mapC);
        //车次统计RDD
        JavaRDD<VehicleDrivingNumberInAreaDto> saveHive = getVehicleDrivingNumberInArea(mapRDD);
        //缓存车次统计
        JavaRDD<VehicleDrivingNumberInAreaDto> areaRDDCache = saveHive.persist(StorageLevel.MEMORY_AND_DISK_SER());
        //落盘车次统计
        saveVehicleDrivingNumberInAreaDtoByTypeList(spark,areaRDDCache,day);
        //原始数据处理完成，释放内存
        lines.unpersist();
        areaRDDCache.unpersist();
        jsc.close();
        spark.stop();
    }

    private static SparkSession getSparkSessionByMaster(String master) {
        SparkSession spark = null;
        if(master == "local" || "local".equals(master)) {
             spark = SparkSession.builder()
                    .appName("VehiclePassTimesStatistics")
                    .master(master)
                    .config("spark.mongodb.output.uri",
                            PropertiesUtil.getProperties("spark.mongodb.output.uri"))
                    .config("spark.serializer", PropertiesUtil.getProperties("spark.serializer.type"))
                    .config("spark.kryoserializer.buffer.max", PropertiesUtil.getProperties("spark.serializer.max.buffer.value"))
                    .config("spark.sql.warehouse.dir", "hdfs://platform-010-030-050-032:8020/apps/hive/warehouse")
                    .config("spark.sql.hive.metastore.jars", "maven")
                    .enableHiveSupport()
                    .getOrCreate();
        }else if(master == "yarn" || "yarn".equals(master)){
            spark = SparkSession.builder()
                    .appName("VehiclePassTimesStatistics")
                    .config("spark.mongodb.output.uri",
                            PropertiesUtil.getProperties("spark.mongodb.output.uri"))
                    .config("spark.serializer", PropertiesUtil.getProperties("spark.serializer.type"))
                    .config("spark.kryoserializer.buffer.max", PropertiesUtil.getProperties("spark.serializer.max.buffer.value"))
                    .enableHiveSupport()
                    .getOrCreate();
        }
        return spark;
    }

    private static JavaRDD<VehicleDrivingNumberInGridDto> getProvinceGirdRDD(JavaRDD<LocationDataInfo> lines,
                                                                             int num,
                                                                             String day,
                                                                             Broadcast<HashMap<Long, String>> mapB,
                                                                             Broadcast<HashMap<Integer, String>> mapC) {
        JavaPairRDD<Long, Long> pair1 = lines.mapPartitionsToPair((PairFlatMapFunction<Iterator<LocationDataInfo>, Long, Long>) documentIterator -> {

            List<Tuple2<Long, Long>> list = new ArrayList<Tuple2<Long, Long>>();
            while (documentIterator.hasNext()) {
                LocationDataInfo locationDataInfo = documentIterator.next();
                Long tid = locationDataInfo.getTid();
                HashSet<Tuple2<Long, Long>> set = new HashSet<Tuple2<Long, Long>>();

                // 过滤无效数据
                if (locationDataInfo.getOriginallat()<=0 || locationDataInfo.getOriginallng()<=0){
                    continue;
                }
                long lat = locationDataInfo.getOriginallat();
                long lon = locationDataInfo.getOriginallng();
                long tileIdMin = VehiclePassTimesQuadtree.getTileNumber(lat, lon, 11);
                set.add(new Tuple2<Long, Long>(tileIdMin, tid));
                list.addAll(set);
            }

            return list.iterator();
        }).partitionBy(new TilePartitioner(num));

        System.out.println("6");
        JavaRDD<VehicleDrivingNumberInGridDto> dt1 = pair1.mapPartitions((FlatMapFunction<Iterator<Tuple2<Long, Long>>, VehicleDrivingNumberInGridDto>) tuple2Iterator -> {

            List<VehicleDrivingNumberInGridDto> dtList = new ArrayList<>();

            HashMap<Long,HashSet<HashMap>> map1 = new HashMap<>(); //11
            HashMap<Long,HashSet<HashMap>> map2 = new HashMap<>(); //10
            HashMap<Long,HashSet<HashMap>> map3 = new HashMap<>(); //9

            while (tuple2Iterator.hasNext()){
                Tuple2<Long, Long> tuple2 = tuple2Iterator.next();

                Long key1 = tuple2._1();
                Long value1 = tuple2._2();

                HashSet<HashMap> set1 = null;
                if (map1.get(key1) == null){
                    set1 = new HashSet<>();
                } else {
                    set1 = map1.get(key1);
                }
                set1.add(createMap(value1, 1));
                map1.put(key1, set1);

                Long key2 = VehiclePassTimesQuadtree.getParentTileId(key1);
                HashSet<HashMap> set2 = null;
                if (map2.get(key2) == null){
                    set2 = new HashSet<>();
                } else {
                    set2 = map2.get(key2);
                }
                set2.add(createMap(value1, 1));
                map2.put(key2,set2);

                Long key3 = VehiclePassTimesQuadtree.getParentTileId(key2);
                HashSet<HashMap> set3 = null;
                if (map3.get(key3) == null){
                    set3 = new HashSet<>();
                } else {
                    set3 = map3.get(key3);
                }
                set3.add(createMap(value1, 1));
                map3.put(key3, set3);
            }

            for(Map.Entry<Long, HashSet<HashMap>> e : map3.entrySet()){
                Long k = e.getKey();
                HashSet<HashMap> v = e.getValue();

                List<HashMap> list = getChildVidList1(k, map2, map1);
                VehicleDrivingNumberInGridDto gridDto = new VehicleDrivingNumberInGridDto();
                gridDto.setTile(k);
                gridDto.setTimes(v.size());
                gridDto.setMonth(getMonth(day));
                gridDto.setDetail(list);

                int[] zxy = VehiclePassTimesQuadtree.tileIdTozxy(k);

                long lon = VehiclePassTimesQuadtree.tile2lon(zxy[1],zxy[0]);
                long lat = VehiclePassTimesQuadtree.tile2lat(zxy[2],zxy[0]);

                long t15 = VehiclePassTimesQuadtree.getTileNumber(lat, lon, 15);
                String district = mapB.getValue().get(t15);
                if (district!=null && !"".equals(district)){
                    //String cityInfo = mapC.value().get(district);
                    String[] areaArr = district.split(",");
                    //areaArr[city,province]
                    if(areaArr.length == 2){
                        int key = Integer.parseInt(areaArr[0]);
                        String provoinceInfo = mapC.value().get(key);
                        if(provoinceInfo!=null && !"".equals(provoinceInfo)) {
                            String[] provoinceArr = provoinceInfo.split(",")[1].split("\\|");
                            gridDto.setDistrict(Long.parseLong(provoinceArr[0]));
                            gridDto.setName(provoinceArr[1]);
                        }
                    }
                }
                dtList.add(gridDto);
            }
            return dtList.iterator();
        });
        return dt1;
    }

    /**
     * 按城市统计
     * @param lines
     * @param num
     * @param day
     * @param mapB
     * @param mapC
     * @return
     */
    private static JavaRDD<VehicleDrivingNumberInGridDto> getCityGirdRDD(JavaRDD<LocationDataInfo> lines,
                                                                         int num,
                                                                         String day,
                                                                         Broadcast<HashMap<Long, String>> mapB,
                                                                         Broadcast<HashMap<Integer,String>> mapC) {
        JavaPairRDD<Long, Long> pair = lines.mapPartitionsToPair((PairFlatMapFunction<Iterator<LocationDataInfo>, Long, Long>) documentIterator -> {
            List<Tuple2<Long, Long>> list = new ArrayList<Tuple2<Long, Long>>();
            while (documentIterator.hasNext()) {
                LocationDataInfo locationDataInfo = documentIterator.next();
                HashSet<Tuple2<Long, Long>> set = new HashSet<Tuple2<Long, Long>>();
                Long tid = locationDataInfo.getTid();
                int longitude = locationDataInfo.getOriginallng();
                int latitude = locationDataInfo.getOriginallat();
                // 过滤无效数据
                if (latitude <= 0 || longitude <= 0) {
                    continue;
                }
                int lon = locationDataInfo.getOriginallng();
                int lat = locationDataInfo.getOriginallat();
                //对应省市地区key
                long tileIdMin = VehiclePassTimesQuadtree.getTileNumber(lat, lon, 15);
                set.add(new Tuple2<Long, Long>(tileIdMin, tid));
                list.addAll(set);
            }
//            System.out.println("mapPartitionsToPair end");
            return list.iterator();
        }).partitionBy(new TilePartitioner(num));
        System.out.println("4");
        //pair:List<Tuple2<Long,Long>>经纬度对应瓦片ID,tid
        JavaRDD<VehicleDrivingNumberInGridDto> dt = pair.mapPartitions((FlatMapFunction<Iterator<Tuple2<Long, Long>>, VehicleDrivingNumberInGridDto>) tuple2Iterator -> {

//            System.out.println("mapPartitions start");

            List<VehicleDrivingNumberInGridDto> dtList = new ArrayList<>();

            HashMap<Long,HashSet<HashMap>> map1 = new HashMap<>(); //15
            HashMap<Long,HashSet<HashMap>> map2 = new HashMap<>(); //14
            HashMap<Long,HashSet<HashMap>> map3 = new HashMap<>(); //13

            while (tuple2Iterator.hasNext()){
                Tuple2<Long, Long> tuple2 = tuple2Iterator.next();
                //瓦片对应地区ID
                Long key1 = tuple2._1();
                //tid
                Long value1 = tuple2._2();

                HashSet<HashMap> set1 = null;
                if (map1.get(key1) == null){
                    set1 = new HashSet<>();
                } else {
                    set1 = map1.get(key1);
                }
                //createMap(tid,1):HashMap<>:(k:"tid",value:tid)(k:"d",v:1)
                set1.add(createMap(value1, 1));
                //map1<k:瓦片ID，v:set(map)>
                map1.put(key1, set1);
                //获取2级瓦片ID
                Long key2 = VehiclePassTimesQuadtree.getParentTileId(key1);
                HashSet<HashMap> set2 = null;
                if (map2.get(key2) == null){
                    set2 = new HashSet<>();
                } else {
                    set2 = map2.get(key2);
                }
                set2.add(createMap(value1, 1));
                map2.put(key2,set2);

                Long key3 = VehiclePassTimesQuadtree.getParentTileId(key2);
                HashSet<HashMap> set3 = null;
                if (map3.get(key3) == null){
                    set3 = new HashSet<>();
                } else {
                    set3 = map3.get(key3);
                }
                set3.add(createMap(value1, 1));
                map3.put(key3, set3);
            }
//            System.out.println("mapPartitions middle");
            for(Map.Entry<Long, HashSet<HashMap>> e : map3.entrySet()){
                Long k = e.getKey();
                HashSet<HashMap> v = e.getValue();

                List<HashMap> list = getChildVidList(k, map2, map1);
                VehicleDrivingNumberInGridDto gridDto = new VehicleDrivingNumberInGridDto();
                gridDto.setTile(k);
                gridDto.setTimes(v.size());
                gridDto.setMonth(getMonth(day));
                gridDto.setDetail(list);

                int[] zxy = VehiclePassTimesQuadtree.tileIdTozxy(k);

                long lon = VehiclePassTimesQuadtree.tile2lon(zxy[1],zxy[0]);
                long lat = VehiclePassTimesQuadtree.tile2lat(zxy[2],zxy[0]);

                long t15 = VehiclePassTimesQuadtree.getTileNumber(lat, lon, 15);
//                System.out.println("mapB————————————"+mapB.getValue().size());
                String district = mapB.getValue().get(t15);
//                System.out.println("district————————————"+district);

                if (district!=null && !"".equals(district)){
                    //String cityInfo = mapC.value().get(district);
                    String[] areaArr = district.split(",");
                    //areaArr[city,province]
                    if(areaArr.length == 2){
                        int key = Integer.parseInt(areaArr[0]);
                        String cityInfo = mapC.value().get(key);
                        if(cityInfo!=null && !"".equals(cityInfo)) {
                            String[] cityArr = cityInfo.split(",")[0].split("\\|");
                            gridDto.setDistrict(Long.parseLong(cityArr[0]));
                            gridDto.setName(cityArr[1]);
                        }
                    }
                }
                dtList.add(gridDto);
            }
//            System.out.println("mapPartitions end");

            return dtList.iterator();
        });
        return dt;
    }


    private static JavaRDD<VehicleDrivingNumberInAreaDto> getVehicleDrivingNumberInArea(JavaPairRDD<String,String> mapRDD) {

        JavaRDD<VehicleDrivingNumberInAreaDto> saveH = mapRDD.reduceByKey(new Function2<String, String, String>() {
            @Override
            public String call(String v1, String v2) throws Exception {
                //210100|沈阳,13112010201:1000:121029120:0,....,13112010201:1000:121029120:0
                return v1 + "," + v2;
            }
        }).mapPartitions((FlatMapFunction<Iterator<Tuple2<String, String>>, VehicleDrivingNumberInAreaDto>)
                iter -> new Iterator<VehicleDrivingNumberInAreaDto>() {
                    @Override
                    public boolean hasNext() {
                        return iter.hasNext();
                    }

                    @Override
                    public VehicleDrivingNumberInAreaDto next() {
                        Tuple2<String, String> tuple = iter.next();
                        String key = tuple._1;
                        String[] areaArr = key.split("\\|");
                        String areacode = areaArr[0];
                        String areaname = null;
                        int type = 5;
                        if(areaArr.length == 3){
                            areaname = areaArr[1];
                            type = Integer.parseInt(areaArr[2]);
                        }

                        //13112010201:1000:121029120:0,....,13112010201:1000:121029120:0
                        String value = tuple._2;
                        HashMap<String, Map> dataHashMap = new HashMap<>();
                        String[] dataArray = value.split(",");
                        for (String v : dataArray) {
                            // 轮询第一步的rdd
                            // v 形如:13112010201:1000:121029120:0
                            String[] x = v.split(":");
                            if(!dataHashMap.containsKey(v)){
                                Map map = new HashMap();
                                map.put("id", Long.valueOf(x[0]));
                                map.put("m", Long.valueOf(x[1]));
                                map.put("t", 0L);
                                map.put("f", false);
                                dataHashMap.put(v, map);
                            }
                        }

                        Collection<Map> collection =  dataHashMap.values();
                        VehicleDrivingNumberInAreaDto veh = new VehicleDrivingNumberInAreaDto();
                        int size = collection.size();
                        try {
                            veh.setDistrict(Long.parseLong(areacode));
                            veh.setName(areaname);
                            veh.setTimes(size);
                            veh.setType(type);
                            veh.setDetail(dataHashMap);
                        } catch (Exception e) {
                            logger.error("veh parse error", e);
                        }
                        return veh;
                    }

                });

        return saveH;
    }

    private static JavaPairRDD<String,String> getVehicleCrossArea(JavaRDD<LocationDataInfo> lines,
                                                                  Broadcast<HashMap<Long, HashSet<Long>>> mapA,
                                                                  Broadcast<HashMap<Long, String>> mapB,
                                                                  Broadcast<HashMap<Integer,String>> mapC
    ) {
        JavaPairRDD<Long,LocationDataInfo> pairRDD = lines.mapPartitionsToPair((PairFlatMapFunction<Iterator<LocationDataInfo>, Long, LocationDataInfo>) iter
                -> new Iterator<Tuple2<Long, LocationDataInfo>>() {
            @Override
            public boolean hasNext() {
                return iter.hasNext();
            }
            @Override
            public Tuple2<Long, LocationDataInfo> next() {
                LocationDataInfo row = iter.next();
                return new Tuple2<>(row.getTid(),row);
            }
        });
        JavaPairRDD<Long,List<LocationDataInfo>> groupRDD = pairRDD.combineByKey(
                (Function<LocationDataInfo, List<LocationDataInfo>>) point -> Stream.of(point).collect(Collectors.toList()),
                (Function2<List<LocationDataInfo>, LocationDataInfo, List<LocationDataInfo>>) (points, p) ->{
                    points.add(p);
                    return points;
                } ,
                (Function2<List<LocationDataInfo>, List<LocationDataInfo>, List<LocationDataInfo>>) (p1, p2) ->{
                    p1.addAll(p2);
                    return p1;
                }
        );
        JavaPairRDD<Long,List<LocationDataInfo>> sortRDD = groupRDD.mapPartitionsToPair((PairFlatMapFunction<Iterator<Tuple2<Long, List<LocationDataInfo>>>, Long, List<LocationDataInfo>>)
                iter -> new Iterator<Tuple2<Long, List<LocationDataInfo>>>() {
                    @Override
                    public boolean hasNext() {
                        return iter.hasNext();
                    }

                    @Override
                    public Tuple2<Long, List<LocationDataInfo>> next() {
                        Tuple2<Long, List<LocationDataInfo>> tuple = iter.next();
                        long tid = tuple._1;
                        List<LocationDataInfo> list = tuple._2;
                        List<LocationDataInfo> resultList = new ArrayList<>();
                        if(list.size()>0 && list!= null){
                            Collections.sort(list);
                            resultList.addAll(list);
                        }
                        return new Tuple2<>(tid,resultList);
                    }
                });
        JavaPairRDD<String,String> mapRDD = sortRDD.flatMapToPair(new PairFlatMapFunction<Tuple2<Long, List<LocationDataInfo>>, String, String>() {
            @Override
            public Iterator<Tuple2<String, String>> call(Tuple2<Long, List<LocationDataInfo>> longListTuple2) throws Exception {
                long tid = longListTuple2._1;
                List<LocationDataInfo> list = longListTuple2._2;
                List<Tuple2<String,String>> resultList = new ArrayList<>();
                Map tidinserted = new HashMap();
                for(LocationDataInfo locationDataInfo:list){

                    int lon = locationDataInfo.getOriginallng();
                    int lat = locationDataInfo.getOriginallat();
                    // 过滤无效数据
                    if (lat <= 0 || lon <= 0) {
                        continue;
                    }

                    long tileIdMin = VehiclePassTimesQuadtree.getTileNumber(lat, lon, 15);
                    //通过瓦片信息获得市、省编号
                    String value0 = mapB.getValue().get(tileIdMin);
                    if (StringUtils.isEmpty(value0)) {
                        continue;
                    }

                    //获得城市编号
                    int citykey = Integer.parseInt(value0.split(",")[0]);
                    //获取省市对应名称格式:code|cname,pcode|pname
                    String value = mapC.getValue().get(citykey);
                    if (StringUtils.isEmpty(value)) {
                        continue;
                    }
                    //拼接结果value值
                    long m = locationDataInfo.getEcuMileage();
                    String resultValue = tid + ":" + m;
                    String[] areaArr = value.split(",");
                    String ckey = areaArr[0]+"|2";
                    String pkey = areaArr[1]+"|1";
                    if(!tidinserted.containsKey(ckey)){
                        resultList.add(new Tuple2<>(ckey,resultValue));
                        tidinserted.put(ckey,null);
                    }
                    if(!tidinserted.containsKey(pkey)){
                        resultList.add(new Tuple2<>(pkey,resultValue));
                        tidinserted.put(pkey,null);
                    }
                    long tileIdMax = VehiclePassTimesQuadtree.getTileNumber(lat, lon, 13);
                    //获取服务站信息
                    HashSet<Long> value1 = mapA.getValue().get(tileIdMax);
                    if (value1 != null && !value1.isEmpty()) {
                        Iterator<Long> it = value1.iterator();
                        while (it.hasNext()) {
                            String skey = String.valueOf(it.next());
                            if(!tidinserted.containsKey(skey)){
                                resultList.add(new Tuple2<>(skey+"|null|5",resultValue));
                                tidinserted.put(skey,null);
                            }
                        }
                    }
                }
                return resultList.iterator();
            }
        });
        return mapRDD;
    }


    private static List<HashMap> getChildVidList(Long parent, HashMap<Long,HashSet<HashMap>> map2, HashMap<Long,HashSet<HashMap>> map1) {
        List<HashMap> list = new ArrayList<>();
        int[] zxy = VehiclePassTimesQuadtree.tileIdTozxy(parent);
        if (zxy[0] == 15) {
            return list;
        }

        long tileIdLt = VehiclePassTimesQuadtree.xyzToTileId(zxy[0] + 1, 2 * zxy[1], 2 * zxy[2]);
        long tileIdRt = VehiclePassTimesQuadtree.xyzToTileId(zxy[0] + 1, 2 * zxy[1] + 1, 2 * zxy[2]);
        long tileIdLb = VehiclePassTimesQuadtree.xyzToTileId(zxy[0] + 1, 2 * zxy[1], 2 * zxy[2] + 1);
        long tileIdRb = VehiclePassTimesQuadtree.xyzToTileId(zxy[0] + 1, 2 * zxy[1] + 1, 2 * zxy[2] + 1);

        HashSet<HashMap> ltSet = new HashSet<>();
        HashSet<HashMap> rtSet = new HashSet<>();
        HashSet<HashMap> lbSet = new HashSet<>();
        HashSet<HashMap> rbSet = new HashSet<>();

        if (zxy[0] == 13) {
            if (map2.get(tileIdLt) != null)
                ltSet = map2.get(tileIdLt);
            if (map2.get(tileIdRt) != null)
                rtSet = map2.get(tileIdRt);
            if (map2.get(tileIdLb) != null)
                lbSet = map2.get(tileIdLb);
            if (map2.get(tileIdRb) != null)
                rbSet = map2.get(tileIdRb);
        } else if (zxy[0] == 14) {
            if (map1.get(tileIdLt) != null)
                ltSet = map1.get(tileIdLt);
            if (map1.get(tileIdRt) != null)
                rtSet = map1.get(tileIdRt);
            if (map1.get(tileIdLb) != null)
                lbSet = map1.get(tileIdLb);
            if (map1.get(tileIdRb) != null)
                rbSet = map1.get(tileIdRb);
        }

        HashMap<String, Object> ltMap = new HashMap<>();
        HashMap<String, Object> rtMap = new HashMap<>();
        HashMap<String, Object> lbMap = new HashMap<>();
        HashMap<String, Object> rbMap = new HashMap<>();

        ltMap.put("tile", tileIdLt);
        ltMap.put("times", ltSet.size());

        rtMap.put("tile", tileIdRt);
        rtMap.put("times", rtSet.size());

        lbMap.put("tile", tileIdLb);
        lbMap.put("times", lbSet.size());

        rbMap.put("tile", tileIdRb);
        rbMap.put("times", rbSet.size());

        if(zxy[0]==14){
            ltMap.put("dl", ltSet);
            lbMap.put("dl", lbSet);
            rtMap.put("dl", rtSet);
            rbMap.put("dl", rbSet);

        }else{
            ltMap.put("dl", getChildVidList(tileIdLt, map2, map1));
            lbMap.put("dl", getChildVidList(tileIdLb, map2, map1));
            rtMap.put("dl", getChildVidList(tileIdRt, map2, map1));
            rbMap.put("dl", getChildVidList(tileIdRb, map2, map1));
        }

        list.add(ltMap);
        list.add(rtMap);
        list.add(lbMap);
        list.add(rbMap);

        return list;
    }
    //
//
    private static List<HashMap> getChildVidList1(Long parent, HashMap<Long,HashSet<HashMap>> map2, HashMap<Long,HashSet<HashMap>> map1) {
        List<HashMap> list = new ArrayList<>();
        int[] zxy = VehiclePassTimesQuadtree.tileIdTozxy(parent);
        if (zxy[0] == 11) {
            return list;
        }

        long tileIdLt = VehiclePassTimesQuadtree.xyzToTileId(zxy[0] + 1, 2 * zxy[1], 2 * zxy[2]);
        long tileIdRt = VehiclePassTimesQuadtree.xyzToTileId(zxy[0] + 1, 2 * zxy[1] + 1, 2 * zxy[2]);
        long tileIdLb = VehiclePassTimesQuadtree.xyzToTileId(zxy[0] + 1, 2 * zxy[1], 2 * zxy[2] + 1);
        long tileIdRb = VehiclePassTimesQuadtree.xyzToTileId(zxy[0] + 1, 2 * zxy[1] + 1, 2 * zxy[2] + 1);

        HashSet<HashMap> ltSet = new HashSet<>();
        HashSet<HashMap> rtSet = new HashSet<>();
        HashSet<HashMap> lbSet = new HashSet<>();
        HashSet<HashMap> rbSet = new HashSet<>();

        if (zxy[0] == 9) {
            if (map2.get(tileIdLt) != null)
                ltSet = map2.get(tileIdLt);
            if (map2.get(tileIdRt) != null)
                rtSet = map2.get(tileIdRt);
            if (map2.get(tileIdLb) != null)
                lbSet = map2.get(tileIdLb);
            if (map2.get(tileIdRb) != null)
                rbSet = map2.get(tileIdRb);
        } else if (zxy[0] == 10) {
            if (map1.get(tileIdLt) != null)
                ltSet = map1.get(tileIdLt);
            if (map1.get(tileIdRt) != null)
                rtSet = map1.get(tileIdRt);
            if (map1.get(tileIdLb) != null)
                lbSet = map1.get(tileIdLb);
            if (map1.get(tileIdRb) != null)
                rbSet = map1.get(tileIdRb);
        }

        HashMap<String, Object> ltMap = new HashMap<>();
        HashMap<String, Object> rtMap = new HashMap<>();
        HashMap<String, Object> lbMap = new HashMap<>();
        HashMap<String, Object> rbMap = new HashMap<>();

        ltMap.put("tile", tileIdLt);
        ltMap.put("times", ltSet.size());

        rtMap.put("tile", tileIdRt);
        rtMap.put("times", rtSet.size());

        lbMap.put("tile", tileIdLb);
        lbMap.put("times", lbSet.size());

        rbMap.put("tile", tileIdRb);
        rbMap.put("times", rbSet.size());

        if(zxy[0]==10){
            ltMap.put("dl", ltSet);
            lbMap.put("dl", lbSet);
            rtMap.put("dl", rtSet);
            rbMap.put("dl", rbSet);
        }else{
            ltMap.put("dl", getChildVidList1(tileIdLt, map2, map1));
            lbMap.put("dl", getChildVidList1(tileIdLb, map2, map1));
            rtMap.put("dl", getChildVidList1(tileIdRt, map2, map1));
            rbMap.put("dl", getChildVidList1(tileIdRb, map2, map1));
        }

        list.add(ltMap);
        list.add(rtMap);
        list.add(lbMap);
        list.add(rbMap);

        return list;
    }

    private static int getMonth(String time) throws ParseException {
        SimpleDateFormat format = new SimpleDateFormat("yyyyMMdd");
        long mills = format.parse(time).getTime();
        Calendar calendar = Calendar.getInstance();
        calendar.setTimeInMillis(mills);
        int year = calendar.get(Calendar.YEAR);
        int month = calendar.get(Calendar.MONTH) + 1;
        int day = calendar.get(Calendar.DAY_OF_MONTH);
        int result = year * 10000 + month * 100;
        if (day <= 10) {
            result += 1;
        } else if (day <= 20) {
            result += 2;
        } else {
            result += 3;
        }
        return result;
    }

    public static HashMap createMap(long tid, int d){
        HashMap m = new HashMap(2){{put("tid", tid);put("d", d);}};
        return m;
    }
    
    public static class LocationDataInfo implements Comparable<LocationDataInfo> {
        private long tid;
        private int originallng;
        private int originallat;
        private long ecuMileage;
        private long cumulativeRunningTime;
        private List<VehicleStatusAdditionDto> statusDataList;
        private List<BreakdownAdditionDto> breakdownList;

        public static LocationDataInfo parseObj(Row row){
            if(row == null){
                return null;
            }
            Long tid = Long.parseLong(row.getAs("tid"));
            int lon = Integer.parseInt(row.getAs("longitude"));
            int lat = Integer.parseInt(row.getAs("latitude"));
            //转换坐标
            double lon1 = new BigDecimal(lon).divide(new BigDecimal(1000000),6,BigDecimal.ROUND_DOWN).doubleValue();
            double lat1 = new BigDecimal(lat).divide(new BigDecimal(1000000),6,BigDecimal.ROUND_DOWN).doubleValue();
            double[] lonlat84 = TransformUtil.gcj02towgs84(lon1,lat1);
            int lon2 = new BigDecimal(lonlat84[0]).multiply(new BigDecimal(1000000)).intValue();
            int lat2 = new BigDecimal(lonlat84[1]).multiply(new BigDecimal(1000000)).intValue();
            //读取ecu里程和累计运行时长，替代附加信息
            long m = (long)Double.parseDouble(row.getAs("ecuMileage"));
            //long c = Long.parseLong(row.getAs("cumulativeRunningTime"));
            //String statusdataInfo = row.getAs("statusaddition");
            //一汽无此列，为null
            //String breakdownInfo = row.getAs("breakdownaddition_breakdownlist");
            //List<VehicleStatusAdditionDto> statusAdditionDtos = VehicleStatusAdditionDto.parseToList(statusdataInfo);
            //List<BreakdownAdditionDto> breakdownAdditionDtos = BreakdownAdditionDto.parseToList(breakdownInfo);

            LocationDataInfo locationDataInfo = new LocationDataInfo();
            locationDataInfo.setTid(tid);
            locationDataInfo.setOriginallat(lat2);
            locationDataInfo.setOriginallng(lon2);
            //添加ecu里程和累计运行时长
            locationDataInfo.setEcuMileage(m);
            locationDataInfo.setCumulativeRunningTime(0L);
            locationDataInfo.setStatusDataList(null);
            locationDataInfo.setBreakdownList(null);
            return locationDataInfo;
        }

        public long getTid() {
            return tid;
        }

        public void setTid(long tid) {
            this.tid = tid;
        }

        public int getOriginallng() {
            return originallng;
        }

        public void setOriginallng(int originallng) {
            this.originallng = originallng;
        }

        public int getOriginallat() { return originallat; }

        public void setOriginallat(int originallat) {
            this.originallat = originallat;
        }

        public long getEcuMileage() {
            return ecuMileage;
        }

        public void setEcuMileage(long ecuMileage) {
            this.ecuMileage = ecuMileage;
        }

        public long getCumulativeRunningTime() {
            return cumulativeRunningTime;
        }

        public void setCumulativeRunningTime(long cumulativeRunningTime) {
            this.cumulativeRunningTime = cumulativeRunningTime;
        }

        public List<VehicleStatusAdditionDto> getStatusDataList() {
            return statusDataList;
        }

        public void setStatusDataList(List<VehicleStatusAdditionDto> statusDataList) {
            this.statusDataList = statusDataList;
        }

        public List<BreakdownAdditionDto> getBreakdownList() {
            return breakdownList;
        }

        public void setBreakdownList(List<BreakdownAdditionDto> breakdownList) {
            this.breakdownList = breakdownList;
        }

        @Override
        public int compareTo(LocationDataInfo o) {
            int result;
            long mileage = o.getEcuMileage() - this.ecuMileage;
            if(mileage > 0){
                result = 1;
            }else if(mileage < 0){
                result = -1 ;
            }else{
                result = 0;
            }
            return result;
        }
    }

    private static AreaTileCache getAreaTile() throws ClassNotFoundException {
        AreaTileCache areaTileCache = new AreaTileCache();

        String driver = "com.mysql.cj.jdbc.Driver"; // URL指向要访问的数据库名game
        String url = PropertiesUtil.getProperties("mysql.url"); // MySQL配置时的用户名
        String user = PropertiesUtil.getProperties("mysql.username"); // MySQL配置时的密码
        String password = PropertiesUtil.getProperties("mysql.password");
        Class.forName(driver);
        try {
            try (
                    Connection conn = DriverManager.getConnection(url, user, password);
                    Statement statement = conn.createStatement();
                    ResultSet rs = statement.executeQuery("SELECT AREA_ID, AREA_TYPE, AREA_DATA, LAST_UPDATE_TIME FROM  lc_area_fence");
            ) {

                while (rs.next()) {
                    Long AREA_ID = rs.getLong("AREA_ID");
                    Integer AREA_TYPE = rs.getInt("AREA_TYPE");
                    String AREA_DATA = rs.getString("AREA_DATA");
                    Long LAST_UPDATE_TIME = rs.getLong("LAST_UPDATE_TIME");
                    List<AreaDataEntity> areaDataEntities = null;
                    try {
                        areaDataEntities = JsonUtil.toList(AREA_DATA, AreaDataEntity.class);
                    } catch (JsonMappingException e) {
                        logger.error(AREA_ID + "——————" + AREA_DATA, e);
                    }
                    AreaEntity areaEntity = new AreaEntity();
                    areaEntity.setAreaType(AREA_TYPE);
                    areaEntity.setLastUpdateTime(LAST_UPDATE_TIME);
                    areaEntity.setAreaId(AREA_ID);
                    areaEntity.setDatas(areaDataEntities);

                    //生成区域与瓦片对应关系的缓存
                    List<Long> list = AreaToTile.areaToTiles(areaEntity, 13);
                    for (Long tileId : list) {
                        areaTileCache.addAreaTile(tileId, areaEntity.getAreaId());
                    }
                }
            }
        } catch (SQLException e) {
            logger.error("SQLException", e);
        } catch (IOException e) {
            logger.error("IOException", e);
        }
        return areaTileCache;
    }


    /**
     * 通用落盘方法，用[MYSQL,HIVE,MONGO]参数，执行对应落盘
     * @param spark
     * @param saveH
     * @param day
     * @param type
     */
    private static void saveVehicleDrivingNumberInAreaDto(SparkSession spark,
                                                                JavaRDD<VehicleDrivingNumberInAreaDto> saveH,
                                                                final String day,
                                                                final String type){

        JavaRDD rowH = saveH.mapPartitions((FlatMapFunction<Iterator<VehicleDrivingNumberInAreaDto>, ? extends Object> ) beans
        -> new Iterator<Object>() {

            @Override
            public boolean hasNext() {
                return beans.hasNext();
            }

            @Override
            public Object next() {
                VehicleDrivingNumberInAreaDto bean = beans.next();
                return VehicleDrivingNumberInAreaDto.DataLoadTypeEnum.valueOf(type).createRow(bean,day);
            }
        });
        //Dataset<Row> VehicleDrivingNumberInAreaDs = spark.createDataFrame(rowH1,VehicleDrivingNumberInAreaDto.DataLoadTypeEnum.valueOf(type).createSchema());
        VehicleDrivingNumberInAreaDto.DataLoadTypeEnum.valueOf(type).save(spark,rowH,day);
    }


    /**
     * 通用落盘方法，用[MYSQL,HIVE,MONGO]参数，执行对应落盘
     * @param spark
     * @param saveH
     * @param day
     * @param type
     */
    private static void saveVehicleDrivingNumberInGridDto(SparkSession spark,
                                                          JavaRDD<VehicleDrivingNumberInGridDto> saveH,
                                                          final String day,
                                                          final String type){

        JavaRDD rowH = saveH.mapPartitions((FlatMapFunction<Iterator<VehicleDrivingNumberInGridDto>, ? extends Object> ) beans
                -> new Iterator<Object>() {

            @Override
            public boolean hasNext() {
                return beans.hasNext();
            }

            @Override
            public Object next() {
                VehicleDrivingNumberInGridDto bean = beans.next();
                return VehicleDrivingNumberInGridDto.DataLoadTypeEnum.valueOf(type).createRow(bean,day);
            }
        });
        //Dataset<Row> VehicleDrivingNumberInAreaDs = spark.createDataFrame(rowH1,VehicleDrivingNumberInAreaDto.DataLoadTypeEnum.valueOf(type).createSchema());
        VehicleDrivingNumberInGridDto.DataLoadTypeEnum.valueOf(type).save(spark,rowH,day);
    }

    private static void saveVehicleDrivingNumberInGridDtoByTypeList(SparkSession spark,
                                                          JavaRDD<VehicleDrivingNumberInGridDto> saveH,
                                                          final String day){
        String types = PropertiesUtil.getProperties("vehicle.save.type");
        String[] typeArr = types.split(",");
        for(String type:typeArr){
            logger.error("瓦片车次统计{}落盘开始...",type);
            long s = System.currentTimeMillis();
            saveVehicleDrivingNumberInGridDto(spark,saveH,day,type);
            logger.error("瓦片车次统计{}落盘结束，共耗时{}ms...",type,System.currentTimeMillis() - s);
        }
    }

    private static void saveVehicleDrivingNumberInAreaDtoByTypeList(SparkSession spark,
                                                                    JavaRDD<VehicleDrivingNumberInAreaDto> saveH,
                                                                    final String day){
        String types = PropertiesUtil.getProperties("vehicle.save.type");
        String[] typeArr = types.split(",");
        for(String type:typeArr){
            logger.error("区域车次统计{}落盘开始...",type);
            long s = System.currentTimeMillis();
            saveVehicleDrivingNumberInAreaDto(spark,saveH,day,type);
            logger.error("区域车次统计{}落盘结束，共耗时{}ms...",type,System.currentTimeMillis() - s);
        }
    }

}
